﻿/**

 * Copyright (c) 2015-2016, FastDev 刘强 (fastdev@163.com) & Quincy.

 *

 * Licensed under the Apache License, Version 2.0 (the "License");

 * you may not use this file except in compliance with the License.

 * You may obtain a copy of the License at

 *

 *      http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

using ConsoleApplication1.Cluster.Entity;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using OF.Notify.DataHost.Cluster.Entity;
using OF.Notify.Master;
using OF.Notify.Entity;
using OF.Notify.Client;
using OF.Notify.DataHost.Cluster;
using OF.DistributeService.Core.Common;
using OF.Notify.DataHost.Cluster.Disk;
using OF.Notify.Channel;


namespace OF.Notify.Test
{
    public static class NodeRemoting
    {
        internal static ConcurrentDictionary<int, DataNode> dataNodeList = new ConcurrentDictionary<int, DataNode>();
        public static Productor productor = null;

        private static bool isInited = false;
        static void Init()
        {
            if (!isInited)
            {
                lock (typeof(NodeRemoting))
                {
                    if (!isInited)
                    {
                        isInited = true;
                        ClusterContext.enableCountApiCall = true;
                        DiskCollectionOnlineMessageStore.EnableDeleteCollectionFile = false;
                        Productor.Init(new MockNotifyServiceProvider());
                    }
                }
            }
        }

        public static void MockNewNode(short indexI, bool autoStart = false)
        {
            Init();
            DataNode dataNode = new DataNode(new DataNodeConfig { 
                    MinRetainStorageSize=104857600,
                    SyncMessageBatchSize=1024,
                    ZookeeperConnection="127.0.0.1:2181",
                    NotifyHostId = (short)indexI,
                    NotifyHostUrl = "http://127.0.0.1:500" + indexI + "/Notify"
            });
            if (autoStart)
            {
                dataNode.Start();
            }
            dataNodeList.AddOrUpdate(indexI, dataNode, (k, v) => dataNode);
        }

        public static void Start()
        {
            foreach (var dataNode in dataNodeList)
            {
                dataNode.Value.Start();
            }
            Thread.Sleep(2000);
            productor = new Productor();
        }

        public static void MockDisposeNode(int indexI)
        {
            DataNode node = null;
            if (!dataNodeList.TryGetValue(indexI, out node))
            {
                throw new Exception("can't find node!");
            }
            node.DoDispose();
        }

        public static int messageI = 0;
        public static bool MockSendMessage(byte topicEnum)
        {
            if (productor == null)
            {
                lock (typeof(NodeRemoting))
                {
                    if (productor == null)
                    {
                        productor = new Productor();
                    }
                }
            }
            TopicMessage topicMessage = new TopicMessage(topicEnum, BitConverter.GetBytes(messageI++));
            Func<int, string[]> func = (loopI) => loopI == 0 ? new string[] { } : null;
            return productor.SendMessage(topicMessage, func);
        }

        public static void Dump(byte topicEnum, string str)
        {
            Util.LogInfo(str);
            foreach (var node in dataNodeList.Values)
            {
                node.Dump(topicEnum);
            }
            Util.LogInfo("");
            Util.LogInfo("");
        }

        public static void DoDispose()
        {
            if (productor != null)
            {
                productor.DoDispose();
                productor = null;

                foreach (var dataNode in dataNodeList)
                {
                    dataNode.Value.DoDispose();
                }

                dataNodeList = new ConcurrentDictionary<int, DataNode>();
            }
        }
    }

}